package rxgo

import (
	"container/list"
	"context"
	"reflect"
	"sync"
	"time"
)

type filterFunc func(context.Context, interface{}) (interface{}, error)

// filtering node implementation of streamOperator
type filteringOperater struct {
	opFunc func(ctx context.Context, o *Observable, in chan interface{}, out chan interface{}) (end bool)
}

func (fop filteringOperater) op(ctx context.Context, o *Observable) {

	in := o.pred.outflow
	out := o.outflow

	// Scheduler
	var wg sync.WaitGroup

	go func() {
		for end := false; !end; { // made panic op re-enter
			end = fop.opFunc(ctx, o, in, out)
		}
		wg.Wait() // waiting all go-routines completed
		o.closeFlow(out)
	}()

}

// Debounce only emit an item from an Observable if a particular timespan has passed without it emitting another item
func (parent *Observable) Debounce(timespan time.Duration) (o *Observable) {
	o = parent.newFilterObservable("debounce")
	o.timespan = timespan
	o.operator = debounceOperator

	return o
}

var debounceOperator = filteringOperater{opFunc: func(ctx context.Context, o *Observable, in chan interface{}, next chan interface{}) (end bool) {
	var latest reflect.Value
	end = false
	go func() {
		for end != true {
			than := time.After(o.timespan)
			<-than
			if latest != reflect.ValueOf(nil) {
				if o.sendToFlow(ctx, latest.Interface(), next) {
					end = true
				}
				latest = reflect.ValueOf(nil)
			}
		}
	}()
	for {
		if end {
			break
		}
		select {
		case <-ctx.Done():
			end = true
		case item, ok := <-in:
			if !ok {
				end = true
			}
			latest = reflect.ValueOf(item)
		}
	}

	return
}}

// Distinct suppress duplicate items emitted by an Observable
func (parent *Observable) Distinct(f filterFunc) (o *Observable) {
	o = parent.newFilterObservable("distinct")
	o.flip = f
	o.operator = distinctOperator

	return o
}

var distinctOperator = filteringOperater{func(ctx context.Context, o *Observable, in chan interface{}, next chan interface{}) (end bool) {
	keyset := make(map[interface{}]interface{})
	var latest reflect.Value
	fv := o.flip.(filterFunc)
	end = false
	for {
		if end {
			break
		}
		select {
		case <-ctx.Done():
			end = true
		case item, ok := <-in:
			if !ok {
				end = true
				break
			}
			latest = reflect.ValueOf(item)
			index, _ := fv(ctx, latest.Interface())
			_, ok2 := keyset[index]
			if !ok2 {
				if o.sendToFlow(ctx, latest.Interface(), next) {
					return
				}
				keyset[index] = index
			}
		}
	}
	return
}}

// ElementAt emit only item n emitted by an Observable
func (parent *Observable) ElementAt(index uint) (o *Observable) {
	o = parent.newFilterObservable("ElementAt")
	o.operator = elementatOperator
	o.index = int(index)
	o.flip = func(x interface{}) bool { return true }
	return o
}

var elementatOperator = filteringOperater{func(ctx context.Context, o *Observable, in chan interface{}, next chan interface{}) (end bool) {
	var latest reflect.Value
	i := 0
	end = false
	for {
		if end {
			break
		}
		select {
		case <-ctx.Done():
			end = true
		case item, ok := <-in:
			if !ok {
				end = true
				break
			}
			latest = reflect.ValueOf(item)

			fv := reflect.ValueOf(o.flip)
			var params = []reflect.Value{latest}
			rs, skip, stop, e := userFuncCall(fv, params)

			var flag interface{} = rs[0].Interface()
			if stop || skip {
				continue
			}
			if e != nil {
				flag = e
			}
			
			if !end {
				if b, ok := flag.(bool); ok && b {
					if i == o.index {
						if o.sendToFlow(ctx, latest.Interface(), next) {
							return
						}
					}
					i++
				}
			}
		}
	}
	return
}}

// First emit only the first item
func (parent *Observable) First() (o *Observable) {
	o = parent.newFilterObservable("First")
	o.operator = firstOperator
	o.flip = func(x interface{}) bool { return true }
	return o
}

var firstOperator = filteringOperater{func(ctx context.Context, o *Observable, in chan interface{}, next chan interface{}) (end bool) {
	var latest reflect.Value
	i := 0
	end = false
	for {
		if end {
			break
		}
		select {
		case <-ctx.Done():
			end = true
		case item, ok := <-in:
			if !ok {
				end = true
				break
			}
			latest = reflect.ValueOf(item)

			fv := reflect.ValueOf(o.flip)
			var params = []reflect.Value{latest}
			rs, skip, stop, e := userFuncCall(fv, params)

			var flag interface{} = rs[0].Interface()
			if stop || skip {
				continue
			}
			if e != nil {
				flag = e
			}
			
			if !end {
				if b, ok := flag.(bool); ok && b {
					if i == 0 {
						if o.sendToFlow(ctx, latest.Interface(), next) {
							return
						}
					}
					i++
				}
			}
		}
	}
	return
}}

// IgnoreElements do not emit any items from an Observable
func (parent *Observable) IgnoreElements() (o *Observable) {
	o = parent.newFilterObservable("IgnoreElements")
	o.operator = ignoreelementsOperator

	return o
}

var ignoreelementsOperator = filteringOperater{func(ctx context.Context, o *Observable, in chan interface{}, next chan interface{}) (end bool) {

	end = false
	for {
		if end {
			break
		}
		select {
		case <-ctx.Done():
			end = true
		case _, ok := <-in:
			if !ok {
				end = true
				break
			}
		}
	}
	return
}}

// Last emit only the last item emitted by an Observable
func (parent *Observable) Last() (o *Observable) {
	o = parent.newFilterObservable("Last")
	o.operator = lastOperator
	o.flip = func(x interface{}) bool { return true }
	return o
}

var lastOperator = filteringOperater{func(ctx context.Context, o *Observable, in chan interface{}, next chan interface{}) (end bool) {
	var latest reflect.Value
	end = false
	for {
		if end {
			break
		}
		select {
		case <-ctx.Done():
			end = true
		case item, ok := <-in:
			if !ok {
				end = true
				break
			}
			var tmp reflect.Value
			tmp = reflect.ValueOf(item)

			fv := reflect.ValueOf(o.flip)
			var params = []reflect.Value{tmp}
			rs, skip, stop, e := userFuncCall(fv, params)

			var flag interface{} = rs[0].Interface()
			if stop || skip {
				continue
			}
			if e != nil {
				flag = e
			}
			
			if !end {
				if b, ok := flag.(bool); ok && b {
					latest = reflect.ValueOf(item)
				}
			}
		}
	}
	if latest != reflect.ValueOf(nil) {
		if o.sendToFlow(ctx, latest.Interface(), next) {
			end = true
		}
	}

	return
}}

// Sample emit the most recent item emitted by an Observable within periodic time intervals
func (parent *Observable) Sample(sample chan interface{}) (o *Observable) {
	o = parent.newFilterObservable("Sample")
	o.operator = sampleOperator
	o.sample = sample
	o.flip = func(x interface{}) bool { return true }
	return o
}

var sampleOperator = filteringOperater{func(ctx context.Context, o *Observable, in chan interface{}, next chan interface{}) (end bool) {
	var latest reflect.Value
	end = false

	for {
		if end {
			break
		}
		select {
		case <-ctx.Done():
			end = true
		case item, ok := <-in:
			if !ok {
				end = true
				break
			}
			var tmp reflect.Value
			tmp = reflect.ValueOf(item)

			fv := reflect.ValueOf(o.flip)
			var params = []reflect.Value{tmp}
			rs, skip, stop, e := userFuncCall(fv, params)

			var flag interface{} = rs[0].Interface()
			if stop || skip {
				continue
			}
			if e != nil {
				flag = e
			}
			
			if !end {
				if b, ok := flag.(bool); ok && b {
					latest = reflect.ValueOf(item)
				}
			}
		case <-o.sample:
			if latest != reflect.ValueOf(nil) {
				if o.sendToFlow(ctx, latest.Interface(), next) {
					end = true
				}
				latest = reflect.ValueOf(nil)
			}
		}
	}
	return
}}

// Skip suppress the first n items emitted by an Observable
func (parent *Observable) Skip(n uint) (o *Observable) {
	o = parent.newFilterObservable("Skip")
	o.operator = skipOperator
	o.index = int(n)
	o.flip = func(x interface{}) bool { return true }
	return o
}

var skipOperator = filteringOperater{func(ctx context.Context, o *Observable, in chan interface{}, next chan interface{}) (end bool) {
	var latest reflect.Value
	i := 0
	end = false
	for {
		if end {
			break
		}
		select {
		case <-ctx.Done():
			end = true
		case item, ok := <-in:
			if !ok {
				end = true
				break
			}
			latest = reflect.ValueOf(item)

			fv := reflect.ValueOf(o.flip)
			var params = []reflect.Value{latest}
			rs, skip, stop, e := userFuncCall(fv, params)

			var flag interface{} = rs[0].Interface()
			if stop || skip {
				continue
			}
			if e != nil {
				flag = e
			}
			
			if !end {
				if b, ok := flag.(bool); ok && b {
					if i < o.index {
						i++
					} else {
						if o.sendToFlow(ctx, latest.Interface(), next) {
							return
						}
					}
				}
			}
		}
	}
	return
}}

// SkipLast suppress the last n items emitted by an Observable
func (parent *Observable) SkipLast(n uint) (o *Observable) {
	o = parent.newFilterObservable("SkipLast")
	o.operator = skiplastOperator
	o.index = int(n)
	o.flip = func(x interface{}) bool { return true }
	return o
}

var skiplastOperator = filteringOperater{func(ctx context.Context, o *Observable, in chan interface{}, next chan interface{}) (end bool) {
	var latest reflect.Value
	tmpList := list.New()
	end = false
	for {
		if end {
			break
		}
		select {
		case <-ctx.Done():
			end = true
		case item, ok := <-in:
			if !ok {
				end = true
				break
			}
			latest = reflect.ValueOf(item)

			fv := reflect.ValueOf(o.flip)
			var params = []reflect.Value{latest}
			rs, skip, stop, e := userFuncCall(fv, params)

			var flag interface{} = rs[0].Interface()
			if stop || skip {
				continue
			}
			if e != nil {
				flag = e
			}
			
			if !end {
				if b, ok := flag.(bool); ok && b {
					tmpList.PushBack(latest.Interface())
				}
			}
			
		}
	}
	len := tmpList.Len()
	count := 0
	for i := tmpList.Front(); i != nil && count < len-o.index; i = i.Next() {
		o.sendToFlow(ctx, i.Value, next)
		count++
	}
	return
}}

// Take emit only the first n items emitted by an Observable
func (parent *Observable) Take(n uint) (o *Observable) {
	o = parent.newFilterObservable("Take")
	o.operator = takeOperator
	o.index = int(n)
	o.flip = func(x interface{}) bool { return true }
	return o
}

var takeOperator = filteringOperater{func(ctx context.Context, o *Observable, in chan interface{}, next chan interface{}) (end bool) {
	var latest reflect.Value
	i := 0
	end = false
	for {
		if end {
			break
		}
		select {
		case <-ctx.Done():
			end = true
		case item, ok := <-in:
			if !ok {
				end = true
				break
			}
			latest = reflect.ValueOf(item)

			fv := reflect.ValueOf(o.flip)
			var params = []reflect.Value{latest}
			rs, skip, stop, e := userFuncCall(fv, params)

			var flag interface{} = rs[0].Interface()
			if stop || skip {
				continue
			}
			if e != nil {
				flag = e
			}
			
			if !end {
				if b, ok := flag.(bool); ok && b {
					if i < o.index {
						i++
						if o.sendToFlow(ctx, latest.Interface(), next) {
							return
						}
					}
				}
			}
		}
	}

	return
}}

// TakeLast emit only the last n items emitted by an Observable
func (parent *Observable) TakeLast(n uint) (o *Observable) {
	o = parent.newFilterObservable("TakeLast")
	o.operator = takelastOperator
	o.index = int(n)
	o.flip = func(x interface{}) bool { return true }
	return o
}

var takelastOperator = filteringOperater{opFunc: func(ctx context.Context, o *Observable, in chan interface{}, next chan interface{}) (end bool) {
	var latest reflect.Value
	tmpList := list.New()
	end = false
	for {
		if end {
			break
		}
		select {
		case <-ctx.Done():
			end = true
		case item, ok := <-in:
			if !ok {
				end = true
				break
			}
			latest = reflect.ValueOf(item)

			fv := reflect.ValueOf(o.flip)
			var params = []reflect.Value{latest}
			rs, skip, stop, e := userFuncCall(fv, params)

			var flag interface{} = rs[0].Interface()
			if stop || skip {
				continue
			}
			if e != nil {
				flag = e
			}
			
			if !end {
				if b, ok := flag.(bool); ok && b {
					if tmpList.Len() < o.index {
						tmpList.PushBack(latest.Interface())
					} else {
						element := tmpList.Front()
						tmpList.Remove(element)
						tmpList.PushBack(latest.Interface())
					}
				}
			}
		}

	}

	for tmpList.Front() != nil {
		if o.sendToFlow(ctx, tmpList.Front().Value, next) {
			return
		}
		element := tmpList.Front()
		tmpList.Remove(element)
	}

	return
}}

func (parent *Observable) newFilterObservable(name string) (o *Observable) {
	//new Observable
	o = newObservable()
	o.Name = name

	//chain Observables
	parent.next = o
	o.pred = parent
	o.root = parent.root

	return o
}
